package com.rd.es.service;

import com.alibaba.fastjson.JSONObject;
import com.rd.es.bean.Rdr_Search;
import com.rd.es.dao.Rdr_SearchDao;
import com.rd.es.utils.MyEsConfig;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Scope;
import org.springframework.web.bind.annotation.RequestMapping;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.function.BiConsumer;

/**
 * @author zhengzhuangmin
 * @date 2022/6/4 00:06
 */

//主要是这个类对不对？
@Slf4j
public class RdrSearchService {

    private static final String RDR_INDEX = "rdr_test";

    @Resource
    private Rdr_SearchDao rdr_searchDao;

    @Resource
    private MyEsConfig esClient;

    @Autowired
    private BulkProcessor bulkProcessor;

    @PreDestroy
    public void destory(){
        try {
            esClient.client().close();
            log.info("esClient客户端已经关闭:{}", esClient.client());
        } catch (IOException e) {
            log.error("关闭restHighLevelClient异常：", e);
        }
    }

    @Bean
    @Scope("prototype") //多例
    public BulkProcessor getBulkAsyncProcessor(){
        RestHighLevelClient restHighLevelClient = esClient.client();
        BiConsumer<BulkRequest, ActionListener<BulkResponse>> consumer =
                (bulkRequest, bulkResponseActionListener) -> restHighLevelClient.bulkAsync(
                        bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener);
        log.info("getBulkAsyncProcessor 中 ES客户端地址:{}", restHighLevelClient);

        return BulkProcessor.builder(consumer, new BulkProcessor.Listener() {
                    @Override
                    public void beforeBulk(long executionId, BulkRequest bulkRequest) {
                        //重写beforeBulk,在每次bulk request发出前执行,在这个方法里面可以知道在本次批量操作中有多少操作数
                        int numberOfActions = bulkRequest.numberOfActions();
                        log.info("同步数量 Executing bulk [{}] with {} requests", executionId, numberOfActions);
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest bulkRequest, BulkResponse bulkResponse) {
                        //重写afterBulk方法，每次批量请求结束后执行，可以在这里知道是否有错误发生。
                        if (bulkResponse.hasFailures()) {
                            log.error("Bulk [{}] executed with failures,response = {}", executionId, bulkResponse.buildFailureMessage());
                        } else {
                            log.info("写入成功 Bulk [{}] completed in {} milliseconds", executionId, bulkResponse.getTook().getMillis());
                            try {
                                restHighLevelClient.close();
                                log.info("运行到最后时的es客户端地址:{}", restHighLevelClient);
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    }

                    @Override
                    public void afterBulk(long l, BulkRequest bulkRequest, Throwable failure) {
                        //重写方法，如果发生错误就会调用。
                        log.error("写入失败 Failed to execute bulk", failure);
                    }
                }).setBulkActions(1000)  //  达到刷新的条数
                .setBulkSize(new ByteSizeValue(5L, ByteSizeUnit.MB)) // 达到 刷新的大小
                .setConcurrentRequests(3) // 并发请求数量, 0不并发, 1并发允许执行
                .setFlushInterval(TimeValue.timeValueSeconds(20))  // 固定刷新的时间频率
                .setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(100L), 3)) // 重试补偿策略
                .build();
    }

    //这样写对吗？如何传参？
    @RequestMapping("/addIndex")
    public void addIndex(Rdr_Search rdr_search) {
        long startTime = System.currentTimeMillis();
        List<Rdr_Search> list = rdr_searchDao.selectAll();
        try {
            if (!list.isEmpty()){
                list.parallelStream().forEach(item->
                        bulkProcessor.add(new IndexRequest(RDR_INDEX).source(rdr_search, XContentType.JSON))
                );
            }
            bulkProcessor.flush();
            bulkProcessor.close();
        }catch (Exception e) {
            log.error("BulkProcessor，插入数据异常", e);
        }
        log.info("use time: " + (System.currentTimeMillis() - startTime) + "ms");
    }

}
